package com.auto.storm;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.util.Map;

import lombok.extern.java.Log;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

/**
 * @ClassName: SocketBolt
 * @Description: 划分产品的blot
 * @author hcw
 * @date 2016年12月27日 上午10:34:57
 * 
 */
@Log
public class SocketBolt extends BaseRichBolt {
	PrintWriter pw;
	private OutputCollector _collector;


	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		this._collector = collector;
	}

	// blot计算单元,把input分词再传个blot
	@Override
	public void execute(Tuple input) {
		String record = input.getString(0);
		log.info("收到的消息:"+record.toString());
		try {
			if (record != null && !record.trim().isEmpty()) {
					 String productID = record.substring(0,2);
					 String productValue= record.substring(3);
					_collector.emit(input, new Values(productID,productValue ));
					log.info("产品编号: productId=" + productID+",产品值:"+productValue);
					//成功通知暂时不用
					_collector.ack(input);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("productID", "productValue"));
	}
}
